package io.kiki.stack.netty.protocol;


import io.kiki.stack.netty.RpcConfigManager;
import io.kiki.stack.netty.channelManager.ChannelContext;
import io.kiki.stack.netty.command.*;
import io.kiki.stack.netty.invoker.RpcCommandType;
import io.kiki.stack.netty.processor.*;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;


@Slf4j
@Setter
@Getter
@Sharable
public class RpcCommandHandler implements CommandHandler {


    ProcessorManager processorManager;

    CommandFactory commandFactory;


    public RpcCommandHandler(CommandFactory commandFactory) {
        this.commandFactory = commandFactory;
        this.processorManager = new ProcessorManager();
        //process request
        this.processorManager.registerProcessor(CommonCommandCode.request, new RequestProcessor(this.commandFactory));
        //process response
        this.processorManager.registerProcessor(CommonCommandCode.response, new ResponseProcessor());
        //process heartbeat
        this.processorManager.registerProcessor(HeartbeatCommandCode.heartbeat, new HeartBeatProcessor());

        this.processorManager.registerDefaultProcessor(new AbstractProcessor<Command>() {
            @Override
            public void doProcess(ChannelContext channelContext, Command command) throws Exception {
                log.error("No processor available for command code {}, msgId {}", command.getCommandCode(), command.getId());
            }
        });
    }


    @Override
    public void handleCommand(ChannelContext channelContext, Object msg) throws Exception {
        this.handle(channelContext, msg);
    }

    private void handle(final ChannelContext channelContext, final Object msg) {
        try {
            if (msg instanceof List) {
                final Runnable handleTask = new Runnable() {
                    @Override
                    public void run() {
                        for (final Object m : (List<?>) msg) {
                            RpcCommandHandler.this.process(channelContext.clone(), m);
                        }
                    }
                };
                if (RpcConfigManager.dispatch_msg_list_in_default_executor()) {
                    // If msg is list ,then the batch submission to biz threadpool can save io thread.
                    // See com.alipay.remoting.decoder.ProtocolDecoder
                    processorManager.getDefaultExecutor().execute(handleTask);
                } else {
                    handleTask.run();
                }
            } else {
                process(channelContext, msg);
            }
        } catch (final Throwable t) {
            processException(channelContext, msg, t);
        }
    }


    private void process(ChannelContext channelContext, Object msg) {
        try {
            final AbstractCommand abstractCommand = (AbstractCommand) msg;
            final Processor processor = processorManager.getProcessor(abstractCommand.getCommandCode());
            processor.process(channelContext, abstractCommand, processorManager.getDefaultExecutor());
        } catch (final Throwable t) {
            processException(channelContext, msg, t);
        }
    }

    private void processException(ChannelContext channelContext, Object msg, Throwable t) {
        if (msg instanceof List) {
            for (final Object m : (List<?>) msg) {
                processExceptionForSingleCommand(channelContext, m, t);
            }
        } else {
            processExceptionForSingleCommand(channelContext, msg, t);
        }
    }

    /*
     * Return error command if necessary.
     */
    private void processExceptionForSingleCommand(ChannelContext channelContext, Object msg, Throwable t) {
        final int id = ((AbstractCommand) msg).getId();
        final String emsg = "Exception caught when processing " + ((msg instanceof RequestCommand) ? "request, id=" : "response, id=");
        log.warn(emsg + id, t);
        if (msg instanceof RequestCommand) {
            final RequestCommand requestCommand = (RequestCommand) msg;
            if (requestCommand.getType() != RpcCommandType.REQUEST_ONEWAY) {
                if (t instanceof RejectedExecutionException) {
                    final ResponseCommand exceptionResponse = this.commandFactory.createExceptionResponse(id, ResponseType.server_threadpool_busy);
                    // RejectedExecutionException here assures no response has been sent back
                    // Other exceptions should be processed where exception was caught, because here we don't known whether ack had been sent back.
                    channelContext.getChannelHandlerContext().writeAndFlush(exceptionResponse).addListener(new ChannelFutureListener() {
                        @Override
                        public void operationComplete(ChannelFuture channelFuture) throws Exception {
                            if (!channelFuture.isSuccess()) {
                                log.error("Write back exception response failed, requestId={}", id, channelFuture.cause());
                            }
                        }

                    });
                }
            }
        }
    }


    @Override
    public void registerProcessor(CommandCode commandCode, Processor processor) {
        this.processorManager.registerProcessor(commandCode, processor);
    }


    @Override
    public void registerDefaultExecutor(Executor executor) {
        this.processorManager.setDefaultExecutor(executor);
    }


    @Override
    public Executor getDefaultExecutor() {
        return this.processorManager.getDefaultExecutor();
    }
}
